package cn.witsky.rtp2ws.push.threadpool;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.LockSupport;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import cn.witsky.rtp2ws.Server;
import cn.witsky.rtp2ws.domain.Rtp;
import cn.witsky.rtp2ws.util.LogUtil;
import cn.witsky.rtp2ws.util.SystemClock;

public class PushTask implements Runnable {
	private static Logger logger = LoggerFactory.getLogger(PushTask.class);
	private volatile boolean finished=false;
	

	public boolean isFinished() {
		return finished;
	}

	long startTime;
	Thread currentThread = null;
	int delayTime = 0;	

	public Rtp rtp;

	public PushTask(Rtp rtp) {
		this.rtp = rtp;
	}

	public Rtp getRtp() {
		return rtp;
	}



	@Override
	public void run() {
		this.currentThread = Thread.currentThread();
		startTime=SystemClock.now();
		long lastSendTime = 0l;
		int readNullcount = 0;		
		this.rtp.setTask(this);
		try {
			int sendSize = 0;
			logger.info("task:{}开始执行了.............", this.rtp.getRtpId());
			int retryCount = rtp.getRetry().getCount();
			int reciveByte =rtp.getRetry().getReciveByte();
			int version=rtp.getRetry().getVersion();
			int reciveByteVersion=rtp.getRetry().getReciveByteVersion();
			int waitTime =0;
			while (true) {
				if(SystemClock.now()-this.startTime>Server.MAX_RUNNING_TIME) {
					logger.warn("task:{}运行超过最大时间了,终止线程",this.rtp.getRtpId());
					break;
				}	
				if (this.rtp.isFinished()) {
					this.rtp.sendEndMsg();
					logger.info("task:{}执行结束了", this.rtp.getRtpId());
					break;
				}
//				if (retryCount > 3) {
//					logger.warn("task:重连已经三次了,不再推送数据",this.rtp.getRtpId());
//					break;
//				}
				int useTime =this.rtp.initWsClient();	
				if(useTime==-1) {
					LogUtil.warn(rtp, "PushTask[run:64]", "task:{}获取连接失败了.......", this.rtp.getRtpId());
					this.waitNanos(100);
					continue;
				}				
				int newRtryCount = rtp.getRetry().getCount();
				if (retryCount < newRtryCount) {
					logger.warn("task:{}又断开了,等待接收recive-byte",this.rtp.getRtpId());
					retryCount =newRtryCount;
					continue;
				}else {	
					reciveByte =rtp.getRetry().getReciveByte();
					version=rtp.getRetry().getVersion();
					reciveByteVersion=rtp.getRetry().getReciveByteVersion();
					if(reciveByte==-1) {	
						this.waitNanos(500);
						boolean suc=this.rtp.getRetry().setReciveByte(0, reciveByteVersion);
						if(suc) {
							logger.warn("task:{}超过了500毫秒没有响应recive-byte,将recive-byte设置为0", this.rtp.getRtpId());
						}else {
							logger.warn("task:{}已经被其它线程修改了",this.rtp.getRtpId());
							continue;
						}
					}
				}				
				if(sendSize<=0) {
					readNullcount++;
					this.waitNanos(this.rtp.getWaittime());
					this.rtp.setReadNullSumCount();
					this.rtp.setReadNullMaxCount(readNullcount);
				}else {
					readNullcount = 0;
					if(sendSize>Server.getOnceSendBytelen()) {
						this.waitNanos(5);
					}else {
						this.waitNanos(this.rtp.getWaittime());
					}
					logger.debug("task:{} waitTime:{} sendSize:{}",this.rtp.getRtpId(),waitTime,sendSize);
				}
				
//				if (sendSize>-1) {
//					readNullcount = 0;
//					waitTime =this.rtp.getWaittime();
//					this.waitNanos(waitTime);
//					logger.debug("task:{} waitTime:{} sendSize:{}",this.rtp.getRtpId(),waitTime,sendSize);
//				} else {
//					readNullcount++;
//					this.waitNanos(5);
//					this.rtp.setReadNullSumCount();
//					this.rtp.setReadNullMaxCount(readNullcount);
//				}
				if (lastSendTime == 0) {
					lastSendTime = SystemClock.now();
				}
				try {
					sendSize = this.rtp.sendMsg(retryCount, version);
				}catch(Exception ex) {
					logger.error("执行发送异常了",ex);
				}
				if (sendSize>-1) {
					long now = SystemClock.now();
					delayTime = (int) (now - lastSendTime);
					lastSendTime = now;
				}
				logger.debug("task:{} sendSize:{} seq:{} delayTime:{}",this.rtp.getRtpId(),sendSize,this.rtp.getSendSeq(),delayTime);
			}
		} catch (Exception ex) {
			logger.error("task:" + this.rtp.getRtpId() + "执行异常了", ex);
		}finally {
			finished=true;
		}
	}

	public void waitNanos(int times) {
		LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(times));
	}

//	public void wakeUp() {
//		if (currentThread != null) {
//			LockSupport.unpark(currentThread);
//		}
//
//	}
	public Thread getCurrentThread() {
		return currentThread;
	}
}
